package cn.doitedu.olap_agg;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import org.apache.flink.util.Collector;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.entity.StringEntity;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClientBuilder;
import org.apache.http.util.EntityUtils;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/3/18
 * @Desc: 学大数据，上多易教育

测试数据：

{"user_id":3,"username":"windy","session_id":"s10","event_id":"search","event_time":1670596213000,"lat":38.089969323508726,"lng":114.35731900345093,"release_channel":"华为应用市场","device_type":"mi8","properties":{"keyword":"usb 移动固态","search_id":"sc01"},"register_phone":"18061581848","user_status":1,"register_time":"2018-08-03 16:46:38","register_gender":1,"register_birthday":"2002-03-06","register_province":"上海","register_city":"上海","register_job":"程序员","register_source_type":1,"gps_province":"河北省","gps_city":"石家庄市","gps_region":"鹿泉区","page_type":"文章页","page_service":"内容服务"}
{"user_id":3,"username":"windy","session_id":"s10","event_id":"search_return","event_time":1670596214000,"lat":38.089969323508726,"lng":114.35731900345093,"release_channel":"华为应用市场","device_type":"mi8","properties":{"keyword":"usb 移动固态","res_cnt":276,"search_id":"sc01"},"register_phone":"18061581848","user_status":1,"register_time":"2018-08-03 16:46:38","register_gender":1,"register_birthday":"2002-03-06","register_province":"上海","register_city":"上海","register_job":"程序员","register_source_type":1,"gps_province":"河北省","gps_city":"石家庄市","gps_region":"鹿泉区","page_type":"文章页","page_service":"内容服务"}
{"user_id":3,"username":"windy","session_id":"s10","event_id":"search_click","event_time":1670596216000,"lat":38.089969323508726,"lng":114.35731900345093,"release_channel":"华为应用市场","device_type":"mi8","properties":{"keyword":"usb 移动固态","search_id":"sc01","item_seq":1,"item_attr":"ad"},"register_phone":"18061581848","user_status":1,"register_time":"2018-08-03 16:46:38","register_gender":1,"register_birthday":"2002-03-06","register_province":"上海","register_city":"上海","register_job":"程序员","register_source_type":1,"gps_province":"河北省","gps_city":"石家庄市","gps_region":"鹿泉区","page_type":"文章页","page_service":"内容服务"}
{"user_id":3,"username":"windy","session_id":"s10","event_id":"search_click","event_time":1670596217000,"lat":38.089969323508726,"lng":114.35731900345093,"release_channel":"华为应用市场","device_type":"mi8","properties":{"keyword":"usb 移动固态","search_id":"sc01","item_seq":1,"item_attr":"ad"},"register_phone":"18061581848","user_status":1,"register_time":"2018-08-03 16:46:38","register_gender":1,"register_birthday":"2002-03-06","register_province":"上海","register_city":"上海","register_job":"程序员","register_source_type":1,"gps_province":"河北省","gps_city":"石家庄市","gps_region":"鹿泉区","page_type":"文章页","page_service":"内容服务"}
{"user_id":3,"username":"windy","session_id":"s10","event_id":"search_click","event_time":1670596218000,"lat":38.089969323508726,"lng":114.35731900345093,"release_channel":"华为应用市场","device_type":"mi8","properties":{"keyword":"usb 移动固态","search_id":"sc01","item_seq":1,"item_attr":"ad"},"register_phone":"18061581848","user_status":1,"register_time":"2018-08-03 16:46:38","register_gender":1,"register_birthday":"2002-03-06","register_province":"上海","register_city":"上海","register_job":"程序员","register_source_type":1,"gps_province":"河北省","gps_city":"石家庄市","gps_region":"鹿泉区","page_type":"文章页","page_service":"内容服务"}



计算逻辑 ：
-- 先从行为事件明细中，过滤出搜索行为分析相关的事件

-- 然后，对这些数据，统一表字段
user_id  search_id      event_id       action_time     key_word     返回条数   点击item_id   点击序号
3,       sc01,           search       ,  t1,            咖啡  ,       \N           \N          \N
3,       sc01,           search_return,  t2,            咖啡  ,      380           \N          \N
3,       sc01,           search_click,   t3,            咖啡  ,       \N      ,  item01  ,     seq01
3,       sc01,           search_click,   t3,            咖啡  ,       \N      ,  item31  ,     seq05

-- 然后，按照相同用户、相同搜索id、相同搜索词，聚合出如下结果：
user_id  search_id   key_word    发起时间,  返回的结果条数，   点击的次数
3,        sc01,       ,咖啡,        t1           380              2

**/
public class Job04_SearchAnalysisOlapAgg {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.setParallelism(1);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        // 建表，映射kafka中的dwd层 行为明细
        tenv.executeSql(
                " CREATE TABLE dwd_user_action_kafka (                            "+
                        " 	release_channel STRING,                                       "+
                        " 	device_type STRING,                                           "+
                        " 	session_id STRING,                                            "+
                        " 	event_id STRING,                                              "+
                        " 	action_time BIGINT,                                            "+
                        " 	properties MAP<STRING,STRING>,                                "+
                        "   user_id bigint,                                               "+
                        "   province string,                                              "+
                        " 	city string,                                                  "+
                        " 	region string,                       				          "+
                        "   page_type string,                                             "+
                        " 	page_business string,                                         "+
                        " 	page_channel string,                                          "+
                        " 	keyword as properties['keyword'],                             "+
                        " 	search_id as properties['search_id'],                         "+
                        " 	res_cnt as cast(properties['res_cnt'] as int),                "+
                        " 	item_seq as cast(properties['item_seq'] as int),              "+
                        " 	rt as to_timestamp_ltz(action_time,3) ,                       "+
                        " 	watermark for rt as rt,                                       "+
                        " 	pt as proctime()                                              "+
                        " ) WITH (                                                        "+
                        "     'connector' = 'kafka',                                      "+
                        "     'topic' = 'dwd-user-action',                                "+
                        "     'properties.bootstrap.servers' = 'doitedu:9092',            "+
                        "     'properties.group.id' = 'g008',                             "+
                        "     'scan.startup.mode' = 'latest-offset',                      "+
                        "     'value.format' = 'json',                                    "+
                        "     'value.fields-include' = 'EXCEPT_KEY'                       "+
                        " )				                                                  "
        );


        // 过滤
        tenv.executeSql("create temporary view filtered_view as " +
                "select * from dwd_user_action_kafka where event_id in ('search','search_return','search_click')");



        // 分组聚合
        tenv.executeSql(
                        " create temporary view agg_view as SELECT                                "+
                        "      release_channel ,                                                  "+
                        "      device_type ,                                                      "+
                        "      session_id ,                                                       "+
                        "      user_id ,                                                          "+
                        "      province ,                                                         "+
                        "      city ,                                                             "+
                        "      region ,                       	                                  "+
                        "      page_type ,                                                        "+
                        "      page_business ,                                                    "+
                        "      page_channel ,                                                     "+
                        "      keyword ,                                                          "+
                        "      '' as splitWords ,                                               "+
                        "      '' as similarWord ,                                              "+
                        "      search_id,                                                         "+
                        "      min(action_time) AS search_start_time,	                          "+
                        "      min(res_cnt) AS res_cnt ,                                          "+
                        "      count(event_id) filter(where event_id='search_click') AS click_cnt "+
                        " FROM TABLE(                                                             "+
                        "     TUMBLE(TABLE filtered_view,DESCRIPTOR(rt),INTERVAL '1' MINUTE)      "+
                        " )                                                                       "+
                        " GROUP BY                                                                "+
                        "      window_start,                                                      "+
                        "      window_end,	                                                      "+
                        "      release_channel ,                                                  "+
                        "      device_type ,                                                      "+
                        "      session_id ,                                                       "+
                        "      user_id ,                                                          "+
                        "      province ,                                                         "+
                        "      city ,                                                             "+
                        "      region ,                       	                                  "+
                        "      page_type ,                                                        "+
                        "      page_business ,                                                    "+
                        "      page_channel ,                                                     "+
                        "      keyword ,                                                          "+
                        "      search_id                                                          "
        );


        // 表转流
        Table table = tenv.from("agg_view");
        DataStream<SearchAggBean> stream = tenv.toDataStream(table, SearchAggBean.class);

        // 出于请求http比较耗时的原因，我们可以对数据流按照  keyword来keyby
        // 这样一来，相同的keyword总会发给同一个subTask，而这个subTask请求过这个词的服务后，就可以把结果缓存在状态中
        DataStream<SearchAggBean> resultStream = stream.keyBy(SearchAggBean::getKeyword)
                .process(new KeyedProcessFunction<String, SearchAggBean, SearchAggBean>() {

                    HttpPost post;
                    JSONObject requestParam;
                    CloseableHttpClient client;

                    MapState<String, Tuple2<String, String>> mapState;

                    @Override
                    public void open(Configuration parameters) throws Exception {

                        client = HttpClientBuilder.create().build();

                        post = new HttpPost("http://doitedu:8081/api/post/simwords");
                        post.addHeader("Content-Type", "application/json;utf-8");
                        post.addHeader("Accept", "application/json;utf-8");

                        requestParam = new JSONObject();


                        TypeInformation<String> typeInfoKey = TypeInformation.of(String.class);
                        TypeInformation<Tuple2<String, String>> typeInfoValue = TypeInformation.of(new TypeHint<Tuple2<String, String>>() {});


                        MapStateDescriptor<String, Tuple2<String, String>> desc = new MapStateDescriptor<>("keyword", typeInfoKey, typeInfoValue);
                        desc.enableTimeToLive(StateTtlConfig.newBuilder(Time.hours(1)).updateTtlOnReadAndWrite().build());

                        mapState = getRuntimeContext().getMapState(desc);


                    }

                    @Override
                    public void processElement(SearchAggBean bean, KeyedProcessFunction<String, SearchAggBean, SearchAggBean>.Context ctx, Collector<SearchAggBean> out) throws Exception {

                        String keyword = bean.getKeyword();
                        String splitWords;
                        String similarWord;


                        Tuple2<String, String> splitAndSimilar = mapState.get(keyword);
                        if(splitAndSimilar != null ){
                            splitWords = splitAndSimilar.f0;
                            similarWord = splitAndSimilar.f1;
                        }else{

                            requestParam.put("origin", keyword);
                            post.setEntity(new StringEntity(requestParam.toJSONString(), "utf-8"));

                            // 发送http请求
                            CloseableHttpResponse response = client.execute(post);

                            String responseJson = EntityUtils.toString(response.getEntity());

                            JSONObject jsonObject = JSON.parseObject(responseJson);
                            splitWords = jsonObject.getString("words");
                            similarWord = jsonObject.getString("similarWord");

                            // 将请求结果，放入状态
                            mapState.put(keyword,Tuple2.of(splitWords,similarWord));

                        }



                        bean.setSplitWords(splitWords);
                        bean.setSimilarWord(similarWord);

                        out.collect(bean);

                    }
                });

        // 流转表
        tenv.createTemporaryView("res",resultStream);

        // 创建一个映射表，映射doris中的结果物理表
        tenv.executeSql(
                        " CREATE TABLE doris_sink(                             "+
                        "     release_channel   string,                        "+
                        "     device_type       string,                        "+
                        "     session_id        string,                        "+
                        "     user_id           bigint,                        "+
                        "     province          string,                        "+
                        "     city              string,                        "+
                        "     region            string,                        "+
                        "     page_type         string,                        "+
                        "     page_business     string,                        "+
                        "     page_channel      string,                        "+
                        "     keyword           string,                        "+
                        "     splitWords        string,                        "+
                        "     similarWord       string,                        "+
                        "     search_id         string,                        "+
                        "     search_start_time bigint,                        "+
                        "     res_cnt           bigint,                        "+
                        "     click_cnt         bigint                         "+
                        " ) WITH (                                             "+
                        "    'connector' = 'doris',                            "+
                        "    'fenodes' = 'doitedu:8030',                       "+
                        "    'table.identifier' = 'dws.search_analysis_olap',  "+
                        "    'username' = 'root',                              "+
                        "    'password' = 'root',                              "+
                        "    'sink.label-prefix' = 'doris_label-011'           "+
                        " )                                                    "
        );


        // 写一个insert语句
        tenv.executeSql("insert into doris_sink select * from res");




        env.execute();
    }



    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class SearchAggBean{

        String release_channel ;
        String device_type ;
        String session_id ;
        Long user_id ;
        String province ;
        String city ;
        String region ;
        String page_type ;
        String page_business ;
        String page_channel ;
        String keyword ;
        String splitWords;  // 分词
        String similarWord;  // 近义词
        String search_id;
        Long search_start_time;
        Integer res_cnt ;
        Long click_cn;
    }


}
